package com.surfilter.massdata.spark.task;

import java.util.ArrayList;
import java.util.Date;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;

import org.apache.commons.lang3.StringUtils;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.FlatMapFunction;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.api.java.function.Function2;
import org.apache.spark.api.java.function.PairFunction;
import org.apache.spark.broadcast.Broadcast;
import org.apache.spark.sql.DataFrame;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SQLContext;

import scala.Tuple2;

import com.act.sparkanalyz.service.impl.SparkService.OutQueueEntity;
import com.act.sparkanalyz.task.ISparkTask;
import com.surfilter.massdata.spark.bean.SAN2202;
import com.surfilter.massdata.spark.model.DnsStatConsts;
import com.surfilter.massdata.spark.model.StructIP;
import com.surfilter.massdata.spark.util.CommonUtils;
import com.surfilter.massdata.spark.util.DateUtil;
import com.surfilter.massdata.spark.util.IPDatabase;
import com.surfilter.massdata.spark.util.URLUtil;

/**
 * 关注网站访问量用户IP类型分布
 * 本省网民访问重点网站的访问量IP类型占比情况（本省网民是指CIP在BR2002中存在，省内网站是指DIP在BR2002中存在）
 */
public class WebSiteVisitsIpTypeTask implements ISparkTask{
	
	private static final Log log = LogFactory.getLog(WebSiteVisitsIpTypeTask.class);
	private static final long serialVersionUID = 1L;
	private String outName;
	
	@Override
	public List<OutQueueEntity> execute(Map<String, DataFrame> dataFrames,Map<String, String> commandMap) {
		log.info("==============WebSiteVisitsIpTypeTask begin====================");
		List<OutQueueEntity> outList = new ArrayList<OutQueueEntity>();
		try{
			DataFrame cipDipFrame = dataFrames.get("cip_dip_frame");
			JavaRDD<Row> cipDipRDD = cipDipFrame.toJavaRDD();
			
			SQLContext sqlContext = cipDipFrame.sqlContext();
			
			JavaRDD<Map<String, Object>> domainRDD = getCipDipFilterRDD(cipDipRDD,sqlContext);
			
	        String dayStr = commandMap.get("-d");
	        Date date = DateUtil.getExecDate(DateUtil.getYesterday(), dayStr);
	        
	        JavaRDD<Map<String, Object>> rdd = getRdd(domainRDD);
	        
	        JavaRDD<SAN2202> javaRDD = createRdd(date, rdd);
	    	DataFrame df = sqlContext.createDataFrame(javaRDD, SAN2202.class);
	    	
			outList.add(new OutQueueEntity(outName, df));
			
			CommonUtils.deleteTaskTable("SAN2202", date, "WEBSITE_FOCUS_IPTYPE", 0,"day");
		}catch(Exception e){
			if(log.isErrorEnabled()){
				log.error(e.getMessage(),e);
			}
		}
		return outList;
	}

	@SuppressWarnings({ "serial", "resource" })
	private JavaRDD<Map<String, Object>> getCipDipFilterRDD(JavaRDD<Row> cipDipRDD,SQLContext sqlContext) {
		
		JavaSparkContext sc = new JavaSparkContext(sqlContext.sparkContext());
		final Broadcast<IPDatabase> ipBroadcast = sc.broadcast(IPDatabase.getInstance());
		
		JavaRDD<Map<String,Object>> domainRDD = cipDipRDD.map(new Function<Row, Map<String, Object>>() {

			@Override
			public Map<String, Object> call(Row row) throws Exception {
				String dip = row.getAs("dip");
				String cip = row.getAs("cip");
				IPDatabase ipDatabase = ipBroadcast.value();
				StructIP structDip = ipDatabase.get(dip);
				StructIP structCip = ipDatabase.get(cip);
				Map<String,Object> map = new HashMap<String,Object>();
				if(structDip != null && structCip != null){
					map.put("dns_visit_count",row.getAs("dns_visit_count"));
					map.put("domain",row.getAs("domain"));
					map.put("ip_type", structCip.get(DnsStatConsts.IPExField.IP_TYPE));
				}
				return map;
			}
		});
		return domainRDD;
	}

	@SuppressWarnings("serial")
	private JavaRDD<Map<String, Object>> getRdd(JavaRDD<Map<String,Object>> domainRDD) {
		JavaRDD<Map<String,Object>> rdd = domainRDD.mapToPair(new PairFunction<Map<String,Object>, String,Long>() {

			@Override
			public Tuple2<String, Long> call(Map<String,Object> map) throws Exception {
				try{
					String domain = CommonUtils.valueOf(map.get("domain"));
					String topDomain = URLUtil.getDomainName("http://" + domain);
					if(StringUtils.isNotBlank(topDomain)){
						long count_value = Long.parseLong(CommonUtils.valueOf(map.get("dns_visit_count")));
						
						return new Tuple2<String, Long>(topDomain+"@"+map.get("ip_type"),count_value);
					}
				}catch(Exception e){
					if(log.isErrorEnabled()){
						log.error(e.getMessage(),e);
					}
				}
				return new Tuple2<String, Long>("",0L);
			}
		}).reduceByKey(new Function2<Long,Long,Long>() {
			
			@Override
			public Long call(Long v1, Long v2){
				return v1 + v2;
			}
		}).map(new Function<Tuple2<String,Long>, Map<String,Object>>() {

			@Override
			public Map<String, Object> call(Tuple2<String, Long> tuple){
				Map<String,Object> map = new HashMap<String,Object>();
				try{
					String key = tuple._1;
					if(StringUtils.isNotBlank(key)){
						String[] keyStr = key.split("@");
						map.put("domain",keyStr[0]);
						String ip_type = "";
						if(keyStr.length > 1){
							ip_type = keyStr[1];
						}
						map.put("ip_type",ip_type);
						map.put("count_value", tuple._2);
					}
				}catch(Exception e){
					if(log.isErrorEnabled()){
						log.error(e.getMessage(),e);
					}
				}
				return map;
			}
		});
		
		return rdd;
	}

	@SuppressWarnings("serial")
	private JavaRDD<SAN2202> createRdd(final Date date,JavaRDD<Map<String, Object>> rdd) {
		
		JavaRDD<SAN2202> javaRDD = rdd.mapPartitions(new FlatMapFunction<Iterator<Map<String,Object>>, SAN2202>() {

			@Override
			public Iterable<SAN2202> call(Iterator<Map<String, Object>> it){
				List<SAN2202> list = new ArrayList<SAN2202>();
				try{
					while(it.hasNext()){
						Map<String, Object> map = it.next();
						String domain = CommonUtils.valueOf(map.get("domain"));
						if(StringUtils.isNotBlank(domain)){
		        			SAN2202 sa = new SAN2202();
		            		String ip_type = CommonUtils.valueOf((map.get("ip_type")));
		            		sa.setYear(DateUtil.getCurrentYear(date));
		            		sa.setHalf_year(DateUtil.getHalfYear(date));
		            		sa.setQuarter(DateUtil.getQuarter(date));
		            		sa.setMonth(DateUtil.getCurrentMonth(date));
		            		sa.setWeek(DateUtil.getCurrentWeek(date));
		            		sa.setDay(DateUtil.getCurrentDay(date));
		            		sa.setHour(0);
		            		sa.setSsfl(0);
		            		sa.setBuss_type("WEBSITE_FOCUS_IPTYPE");
		            		sa.setBuss_value(ip_type);
		            		sa.setTop_domain(domain);
		            		sa.setCount_value(Long.parseLong(CommonUtils.valueOf(map.get("count_value"))));
		            		list.add(sa);
		        		}
					}
				}catch(Exception e){
					if(log.isErrorEnabled()){
						log.error(e.getMessage(),e);
					}
				}
				return list;
			}
		});
		return javaRDD;
	}
}
